"""
Case Type   : 权限
Case Name   : show grants--显示访问目录对象directory的权限
Create At   : 2024.8.19
Owner       : wang-tianjing1
Description :
    1、创建用户
    2、修改enable_access_server_director，创建目录对象
    3、授予权限
    4、清理环境
Expect      :
    1、成功
    2、enable_access_server_director为off时合理报错，为on时成功
    3、成功
    4、成功
History     :
    modified by @xuwenfei1 2025-3.19:在执行show grants for user前，将参数b_compatibility_user_host_auth的值设置为true
"""

import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class OpreatorTest(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_primary = CommonSH('PrimaryDbBmode')
        self.user_node = Node('PrimaryDbBmode')
        self.constant = Constant()
        self.db_user = 'u_sg_tester10'
        self.db_directory = 't_directory_0010'
        status = self.sh_primary.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

    def test_show(self):
        text = '----step1:创建用户;expect:成功----'
        self.log.info(text)
        sql_cmd = f'set current_schema to public;' \
                  f'show current_schema;' \
                  f"drop user if exists {self.db_user} cascade;" \
                  f"create user {self.db_user} with sysadmin password" \
                  f"'{macro.COMMON_PASSWD}';"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('SET', sql_res, '执行失败' + text)
        self.assertIn('public', sql_res, '执行失败' + text)
        self.assertIn('DROP ROLE', sql_res, '执行失败' + text)
        self.assertIn('CREATE ROLE', sql_res, '执行失败' + text)

        text = '----step2:===修改enable_access_server_directory = \'off\'，授权创建目录对象;expect:合理报错----'
        self.log.info(text)
        guc1 = self.sh_primary.execute_gsguc('reload',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             "enable_access_server_directory = 'off'")
        restart = self.sh_primary.restart_db_cluster()
        self.assertTrue(restart, '执行失败:' + text)
        self.log.info(guc1)
        sql_cmd1 = f'show enable_access_server_directory;' \
                   f'create or replace directory {self.db_directory} as \'/tmp\';'
        self.log.info(sql_cmd1)
        connect = f'-U {self.db_user} -W {macro.COMMON_PASSWD}'
        sql_res1 = self.sh_primary.execut_db_sql(sql_cmd1, sql_type=connect)
        self.log.info(sql_res1)
        self.assertIn('off', sql_res1, '执行失败' + text)
        self.assertIn('ERROR', sql_res1, '执行失败' + text)
        text = '----修改enable_access_server_directory = \'on\'，再次创建目录对象，except：成功----'
        self.log.info(text)
        guc2 = self.sh_primary.execute_gsguc('reload',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             "enable_access_server_directory = 'on'")
        restart = self.sh_primary.restart_db_cluster()
        self.assertTrue(restart, '执行失败:' + text)
        self.log.info(guc2)
        guc_cmd2 = f'show enable_access_server_directory;'
        self.log.info(guc_cmd2)
        connect = f'-U {self.db_user} -W {macro.COMMON_PASSWD}'
        guc_res2 = self.sh_primary.execut_db_sql(guc_cmd2, sql_type=connect)
        self.log.info(guc_res2)
        self.assertIn('on', guc_res2, '执行失败' + text)
        sql_cmd2 = f'create or replace directory {self.db_directory} as \'/tmp\';'
        self.log.info(sql_cmd2)
        connect = f'-U {self.db_user} -W {macro.COMMON_PASSWD}'
        sql_res2 = self.sh_primary.execut_db_sql(sql_cmd2, sql_type=connect)
        self.log.info(sql_res2)
        self.assertIn('CREATE DIRECTORY', sql_res2, '执行失败' + text)

        text = '----step3:授予权限,expect:成功----'
        self.log.info(text)
        sql_cmd3 = f'show current_schema;'
        self.log.info(sql_cmd3)
        sql_res3 = self.sh_primary.execut_db_sql(sql_cmd3, sql_type=connect)
        self.log.info(sql_res3)
        self.assertIn('(1 row)', sql_res3, '执行失败' + text)

        sql_cmd4 = f'grant read,write,alter,drop on directory {self.db_directory} to {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd4)
        sql_res4 = self.sh_primary.execut_db_sql(sql_cmd4, sql_type=connect)
        self.log.info(sql_res4)
        self.assertIn('GRANT', sql_res4, '执行失败' + text)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res4, '执行失败' + text)

        sql_cmd5 = f'grant read,write,alter,drop on directory {self.db_directory} to {self.db_user} with grant option;' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_directory where dirname = \'{self.db_directory}\';'
        self.log.info(sql_cmd5)
        sql_res5 = self.sh_primary.execut_db_sql(sql_cmd5, sql_type=connect)
        self.log.info(sql_res5)
        self.assertIn('GRANT', sql_res5, '执行失败' + text)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('(1 row)', sql_res5, '执行失败' + text)
        self.assertIn('t_directory_0010', sql_res5, '执行失败' + text)

        sql_cmd6 = f'revoke read,write,alter,drop on directory {self.db_directory} from {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_directory where dirname = \'{self.db_directory}\';'
        self.log.info(sql_cmd6)
        sql_res6 = self.sh_primary.execut_db_sql(sql_cmd6, sql_type=connect)
        self.log.info(sql_res6)
        self.assertIn('REVOKE', sql_res6, '执行失败' + text)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('(1 row)', sql_res6, '执行失败' + text)
        self.assertIn('t_directory_0010', sql_res5, '执行失败' + text)

        sql_cmd7 = f'grant all privileges on directory {self.db_directory} to {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd7)
        sql_res7 = self.sh_primary.execut_db_sql(sql_cmd7, sql_type=connect)
        self.log.info(sql_res7)
        self.assertIn('GRANT', sql_res7, '执行失败' + text)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res7, '执行失败' + text)

        sql_cmd8 = f'grant all privileges on directory {self.db_directory} to {self.db_user} with grant option;' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_directory where dirname = \'{self.db_directory}\';'
        self.log.info(sql_cmd8)
        sql_res8 = self.sh_primary.execut_db_sql(sql_cmd8, sql_type=connect)
        self.log.info(sql_res8)
        self.assertIn('GRANT', sql_res8, '执行失败' + text)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('(1 row)', sql_res8, '执行失败' + text)
        self.assertIn('t_directory_0010', sql_res8, '执行失败' + text)

        sql_cmd9 = f'revoke all privileges on directory {self.db_directory} from {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_directory where dirname = \'{self.db_directory}\';'
        self.log.info(sql_cmd9)
        sql_res9 = self.sh_primary.execut_db_sql(sql_cmd9, sql_type=connect)
        self.log.info(sql_res9)
        self.assertIn('REVOKE', sql_res9, '执行失败' + text)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('(1 row)', sql_res9, '执行失败' + text)
        self.assertIn('t_directory_0010', sql_res8, '执行失败' + text)

    def tearDown(self):
        text = '--step4:清理环境;expect:成功--'
        self.log.info(text)
        sql_cmd = f'drop directory if exists {self.db_directory} ;'
        self.log.info(sql_cmd)
        connect = f'-U {self.db_user} -W {macro.COMMON_PASSWD}'
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, sql_type=connect)
        self.log.info(sql_res)
        sql_clean = f'drop owned by {self.db_user} cascade;' \
                    f"drop user {self.db_user};"
        self.log.info(sql_clean)
        clean_res = self.sh_primary.execut_db_sql(sql_clean)
        self.log.info(clean_res)
        self.assertNotIn("ERROR", sql_res, '执行失败' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
